package com.spark.mooc.ch6_sparksql.part04_fromRDDtoDataFrame

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}

/**
 * @description: 使用编程方式定义RDD模式
 *              事先不知道字段，后面通过动态的方式得到信息
 * @time: 2020/11/29 10:24
 * @author: lhy
 */


object ProgrammingToDefineRDD {
    def main(args: Array[String]): Unit = {
        val spark: SparkSession = SparkSession.builder().appName("ProgrammingToDefineRDD").master("local").getOrCreate()
        import spark.implicits._            // 导入包，支持把一个RDD隐式转换成一个DataFrame
        //生成字段
        val fields: Array[StructField] = Array(StructField("name",StringType,nullable = true),
                                               StructField("age",IntegerType,nullable = true))
        // schema描述了模式信息，模式中包含name和age两个字段 即：schema就是表头
        val schema: StructType = StructType(fields)
        // 下面加载文件生成RDD
        val peopleRDD: RDD[String] = spark.sparkContext.textFile("data.txt/people.txt")
        // 对peopleRDD 这个RDD中的每一行元素都进行解析
        // rowRDD 就是"表中的记录"
        val rowRDD: RDD[Row] = peopleRDD.map(_.split(","))
                                        .map(attributes => Row(attributes(0),attributes(1).trim.toInt))
        // 下面把"表头"和"表中的记录"拼接起来
        val peopleDF: DataFrame = spark.createDataFrame(rowRDD,schema)
        // 必须注册成临时表才能供下面的查询使用
        peopleDF.createOrReplaceTempView("people")
        val results: DataFrame = spark.sql("SELECT name,age FROM people")
        results.show()
        results.map(attributes => "name: "+attributes(0)+","+"age: "+attributes(1)).show()
    }
}
